package com.persagy.iot.func

import com.persagy.iot.app.IOTApp.no_value
import com.persagy.iot.bean.IOTData
import com.persagy.iot.utils.IOTUtils.timestampConverter
import org.apache.flink.api.common.functions.FlatMapFunction
import org.apache.flink.api.common.time.Time
import org.apache.flink.util.Collector

/**
 * 由于每一条数据包含多个功能号，表号
 * 自定义 flatMap 函数
 * @param lastTime 迟到阈值
 * @param loseTime 丢数阈值
 * @param earlyTime 数据产生时间大于入库时间的阈值
 * @param timeOffer 数据产生时间时间差
 */
class SplitData(lastTime: Long, loseTime: Long, earlyTime: Long, timeOffer: Long) extends FlatMapFunction[String, IOTData] {


  override def flatMap(input: String, collector: Collector[IOTData]): Unit = {

    try {
      /* 处理数据，提取 数据上报包 */
      val arr1 = input.split("\t")
      val builds: Array[String] = arr1(3).split("&")
      for (elem <- builds) {

        val arr2: Array[String] = elem.split(";")

        val build: String = arr2(0).toString
        val sign: String = arr2(5).toString
        val eventTimeStr: String = arr2(3)
        val eventTime: Long = timestampConverter("yyyyMMddHHmmss", eventTimeStr, Time.hours(0).toMilliseconds)
        val sysTime: Long = System.currentTimeMillis()
        for (i <- 7 until (arr2.length, 2) if (!no_value.equals(arr2(i + 1)))) {
          val funId: String = arr2(i)
          val value: Double = arr2(i + 1).toDouble
          val status: Int = getStatus(eventTime, sysTime, lastTime, loseTime, earlyTime)

          /* rowKey */
          val rowKey = build + ":" + sign + ":" + funId

          /* 转为 iotData 实体类 */
          val iotData: IOTData = IOTData(rowKey, build, sign, funId, value, eventTimeStr, eventTime, sysTime, 0L, status)

          collector.collect(iotData)
        }
      }
    } catch {
      case ex: Exception => ex.printStackTrace()
    }

  }

  /**
   * 判断该数据状态：
   * 0：正常
   * 1：迟到
   * 2：丢数
   * 3：异常(数据产生时间晚于系统时间大于阈值的状态)
   * @param eventTime 数据产生时间
   * @param sysTime 系统时间
   * @param lastTime 迟到阈值
   * @param loseTime 丢数阈值
   * @param earlyTime 数据产生时间大于入库时间的阈值
   * @param
   */
  def getStatus(eventTime: Long, sysTime: Long, lastTime: Long, loseTime: Long, earlyTime: Long): Int ={

    val var1: Long = sysTime - eventTime
    println("sysTime: " + sysTime + "  -----  eventTime: " + eventTime + "  -----  时间差为: " + var1)
    if (var1 < -earlyTime){
      3
    } else if (var1 > -earlyTime && var1 <= lastTime * 1000){
      0
    } else if (lastTime * 1000 < var1 && var1 <= loseTime * 1000) {
      1
    } else {
      2
    }

  }

}
